package com.ksc.wordcount.task.reduce;

import com.ksc.wordcount.conf.AppConfig;
import com.ksc.wordcount.datasourceapi.PartionWriter;
import com.ksc.wordcount.shuffle.DirectShuffleWriter;
import com.ksc.wordcount.shuffle.ShuffleBlockId;
import com.ksc.wordcount.shuffle.nettyimpl.client.ShuffleClient;
import com.ksc.wordcount.task.KeyValue;
import com.ksc.wordcount.task.Task;
import com.ksc.wordcount.task.TaskStatusEnum;
import com.ksc.wordcount.task.map.MapStatus;


import java.util.UUID;
import java.util.stream.Stream;

public class MiddleReduceTask extends Task<MapStatus> {

    ShuffleBlockId[] shuffleBlockId;
    //String destDir;
    ReduceFunction reduceFunction;
    PartionWriter partionWriter;

    int reduceTaskNum;  // Reduce任务数量


    public MiddleReduceTask(MiddleReduceContext reduceTaskContext) {
        super(reduceTaskContext);
        this.shuffleBlockId = reduceTaskContext.getShuffleBlockId();
        //this.destDir = reduceTaskContext.getDestDir();
        this.reduceTaskNum=reduceTaskContext.getReduceTaskNum();
        this.reduceFunction = reduceTaskContext.getReduceFunction();
        this.partionWriter = reduceTaskContext.getPartionWriter();
    }




    public MapStatus runTask() throws Exception {
        Stream<KeyValue> stream=Stream.empty();
        for(ShuffleBlockId shuffleBlockId:shuffleBlockId){
            stream=Stream.concat(stream,new ShuffleClient().fetchShuffleData(shuffleBlockId));
        }


        Stream sortStream = reduceFunction.reduce(stream);
        String shuffleId = UUID.randomUUID().toString();

        // 将任务执行结果写入Shuffle文件中
        DirectShuffleWriter shuffleWriter =
                new DirectShuffleWriter(AppConfig.shuffleTempDir, shuffleId, applicationId, partionId, reduceTaskNum);

        //6、executor写shuffle数据到本地
        shuffleWriter.write(sortStream);
        shuffleWriter.commit();

        // 返回Shuffle任务的状态
        return shuffleWriter.getMapStatus(taskId);
    }
}
